#include "ORCL.h"

__thread memcached_st *memc = NULL;

void sync_publish_dc_target(struct orcl_context_sync* sync_ctx){
    assert(sync_ctx != NULL);

    assert(sync_ctx->dct_ctx != NULL && sync_ctx->dct_pd != NULL 
        && sync_ctx->dc_target != NULL);

    // key名称
    char machine_name[ORCL_QP_NAME_SIZE]={};
    // 两个参数，一个是节点id，一个是上下文ids
    sprintf(machine_name, "server-sync-dct-%d-%d", sync_ctx->node_id, sync_ctx->orcl_ctx_id);

    int len = strlen(machine_name);

    // 接收缓冲区是message
    struct orcl_dct_attr dct_attr;
    memcpy(dct_attr.name, machine_name, len);
    dct_attr.name[len] = 0;	/* Add the null terminator */
    // 接收缓冲区就是转发的缓冲区
    dct_attr.buf_addr = sync_ctx->message_buffer;
    dct_attr.buf_size = sync_ctx->message_buffer_size;
    // 只有message buf会做接收缓冲区
    dct_attr.rkey =sync_ctx->dct_mes_mr->rkey;
    dct_attr.lid = orcl_get_local_lid(sync_ctx->dc_target->context, sync_ctx->dev_port_id);
    dct_attr.dct_num = sync_ctx->dc_target->dct_num;

    // 获取全局gid
    int ret = ibv_query_gid(sync_ctx->dct_ctx, sync_ctx->dev_port_id, 0, &(dct_attr.gid));

    assert(ret == 0);

    // 将需要的内容共享出去
    orcl_publish(dct_attr.name, &dct_attr, sizeof(orcl_dct_attr));
}


// 发布DCT的基本信息
void orcl_publish_dc_target(struct orcl_context* ctx){
    assert(ctx != NULL);

    // 获取当前节点的组内id
    int inner_group_id = ctx->node_id % ctx->group_size;
    
    int i;
    for(i = 0; i < ctx->group_num; i++){
        // 当前组不需要转发，不需要共享dct
        if(ctx->group_id == i){
            // 跳过
            continue;
        }

        assert(ctx->dct_ctx[i] != NULL && ctx->dct_pd[i] != NULL 
            && ctx->dc_target[i] != NULL);

        // key名称
        char machine_name[ORCL_QP_NAME_SIZE]={};
        // 第一个是转发节点所在的组号，第二个是转发节点的组内id，第三个是能接受转发包的来源组，第四个是上下文编号
        sprintf(machine_name, "server-dct-%d-%d-%d-%d", ctx->group_id, inner_group_id, i, ctx->orcl_ctx_id);

        int len = strlen(machine_name);

        struct orcl_dct_attr dct_attr;
        memcpy(dct_attr.name, machine_name, len);
        dct_attr.name[len] = 0;	/* Add the null terminator */
        // 接收缓冲区就是转发的缓冲区
        dct_attr.buf_addr = ctx->dct_forward_buf[i];
        dct_attr.buf_size = FORWARD_DCT_BUF_SIZE;
        // 只有message buf会做接收缓冲区
        dct_attr.rkey =ctx->dct_forward_mr[i]->rkey;
        dct_attr.lid = orcl_get_local_lid(ctx->dc_target[i]->context, ctx->dev_port_id);
        dct_attr.dct_num = ctx->dc_target[i]->dct_num;

        // 获取gid
        int ret = ibv_query_gid(ctx->dct_ctx[i], ctx->dev_port_id, 0, &(dct_attr.gid));

        assert(ret == 0);

        // 将需要的内容共享出去
        orcl_publish(dct_attr.name, &dct_attr, sizeof(orcl_dct_attr));
    }
}


void orcl_publish_conn_qp(struct orcl_context* ctx){

    assert(ctx != NULL);

    // 发布RC所有的队列
    // 分别包含node_id,ctx_id,要连接的节点的组内id，一共三个东西
    // 另外其也需要包含forward和normal这样的信息
    int i;
    // 首先处理normal，一共要和group_size个节点连接，跳过自己
    for(i = 0; i < ctx->group_size; i++){
        
        if(ctx->node_id % ctx->group_size == i){
            // 跳过与自己的连接
            continue;
        }

        // 查看对应数据结构是不是已经成功初始化
        assert(ctx->rc_normal_ctx[i] != NULL && ctx->rc_normal_pd[i] != NULL 
        && ctx->rc_normal_qp[i] != NULL);

        // 发布让其他节点连接的连接的qp信息
        char machine_name[ORCL_QP_NAME_SIZE]={};
        sprintf(machine_name, "server-normal-%d-%d-%d", ctx->node_id, ctx->orcl_ctx_id, i);

        // 查看其长度
        int len = strlen(machine_name);

        // 构造共享数据，当一个rc作为远端，那么一定是写messagebuf的。
        struct orcl_qp_attr qp_attr;
        memcpy(qp_attr.name, machine_name, len);
        qp_attr.name[len] = 0;	/* Add the null terminator */
        qp_attr.buf_addr = ctx->message_buffer;
        qp_attr.buf_size = ctx->message_buffer_size;
        // 只有message buf会做接收缓冲区
        qp_attr.rkey =ctx->rc_normal_mes_mr[i]->rkey;
        qp_attr.lid = orcl_get_local_lid(ctx->rc_normal_qp[i]->context, ctx->dev_port_id);
        qp_attr.qpn = ctx->rc_normal_qp[i]->qp_num;

        int ret = ibv_query_gid(ctx->rc_normal_ctx[i], ctx->dev_port_id, 0, &(qp_attr.gid));

        assert(ret == 0);

        orcl_publish(qp_attr.name, &qp_attr, sizeof(struct orcl_qp_attr));
    }

    // 这里发布所有的forward队列
    for(i = 0; i < ctx->group_size; i++){
        if(ctx->node_id % ctx->group_size == i){
            // 跳过与自己的连接
            continue;
        }

        // 查看对应数据结构是不是已经成功初始化
        assert(ctx->rc_forward_ctx[i] != NULL && ctx->rc_forward_pd[i]!=NULL 
        && ctx->rc_forward_qp[i] != NULL);

        // 发布让其他节点连接的连接的qp信息
        char machine_name[ORCL_QP_NAME_SIZE]={};
        sprintf(machine_name, "server-forward-%d-%d-%d", ctx->node_id, ctx->orcl_ctx_id, i);

        // 查看其长度
        int len = strlen(machine_name);

        // 构造共享数据，当一个rc作为远端，那么一定是写messagebuf的
        struct orcl_qp_attr qp_attr;
        memcpy(qp_attr.name, machine_name, len);
        qp_attr.name[len] = 0;	/* Add the null terminator */
        qp_attr.buf_addr = ctx->message_buffer;
        qp_attr.buf_size = ctx->message_buffer_size;
        qp_attr.rkey =ctx->rc_forward_mes_mr[i]->rkey;
        qp_attr.lid = orcl_get_local_lid(ctx->rc_forward_qp[i]->context, ctx->dev_port_id);
        qp_attr.qpn = ctx->rc_forward_qp[i]->qp_num;

        int ret = ibv_query_gid(ctx->rc_forward_ctx[i], ctx->dev_port_id, 0, &(qp_attr.gid));

        assert(ret == 0);

        orcl_publish(qp_attr.name, &qp_attr, sizeof(struct orcl_qp_attr));

    }
    
}

// Insert key -> value mapping into memcached running at ORCL_REGISTRY_IP.
void orcl_publish(const char *key, void *value, int len){
    assert(key != NULL && value != NULL && len > 0);
	memcached_return rc;

	if(memc == NULL) {
		memc = orcl_create_memc();
	}

	rc = memcached_set(memc, key, strlen(key), (const char *) value, len, 
		(time_t) 0, (uint32_t) 0);

	if (rc != MEMCACHED_SUCCESS) {
		char *registry_ip = orcl_getenv("ORCL_REGISTRY_IP");
		fprintf(stderr, "\tORCL: Failed to publish key %s. Error %s. "
			"Reg IP = %s\n", key, memcached_strerror(memc, rc), registry_ip);
		exit(-1);
	}
}


/* 
 * Get the value associated with "key" into "value", and return the length
 * of the value. If the key is not found, return NULL and len -1. For all
 * other errors, terminate.
 *
 * This function sometimes gets called in a polling loop - ensure that there
 * are no memory leaks or unterminated memcached connections! We don't need
 * to free() the resul of getenv() since it points to a string in the process
 * environment.
 */
// 获取远程节点信息
int orcl_get_published(const char *key, void **value){
    assert(key != NULL);
	if(memc == NULL) {
		memc = orcl_create_memc();
	}

	memcached_return rc;
	size_t value_length;
	uint32_t flags;

	*value = memcached_get(memc, key, strlen(key), &value_length, &flags, &rc);

	if(rc == MEMCACHED_SUCCESS ) {
		return (int) value_length;
	} else if (rc == MEMCACHED_NOTFOUND) {
		assert(*value == NULL);
		return -1;
	} else {
		char *registry_ip = orcl_getenv("ORCL_REGISTRY_IP");
		fprintf(stderr, "ORCL: Error finding value for key \"%s\": %s. "
			"Reg IP = %s\n", key, memcached_strerror(memc, rc), registry_ip);
		exit(-1);
	}
	
	/* Never reached */
	assert(false);
}

struct orcl_dct_attr* orcl_get_published_dct(const char *dct_name){
    struct orcl_dct_attr *ret;
    assert(dct_name != NULL && strlen(dct_name) < ORCL_QP_NAME_SIZE - 1);
    assert(strstr(dct_name, ORCL_RESERVED_NAME_PREFIX) == NULL);

    int len = strlen(dct_name);
	int i;
	for(i = 0; i < len; i++) {
		if(dct_name[i] == ' ') {
			fprintf(stderr, "ORCL: Space not allowed in DCT name\n");
			exit(-1);
		}
	}

    int ret_len = orcl_get_published(dct_name, (void **) &ret);

    assert(ret_len == sizeof(struct orcl_dct_attr) || ret_len == -1);
	_unused(ret_len);

	return ret;
}


struct orcl_qp_attr* orcl_get_published_qp(const char *qp_name){
    struct orcl_qp_attr *ret;
	assert(qp_name != NULL && strlen(qp_name) < ORCL_QP_NAME_SIZE - 1);
	assert(strstr(qp_name, ORCL_RESERVED_NAME_PREFIX) == NULL);

	int len = strlen(qp_name);
	int i;
	for(i = 0; i < len; i++) {
		if(qp_name[i] == ' ') {
			fprintf(stderr, "ORCL: Space not allowed in QP name\n");
			exit(-1);
		}
	}

	int ret_len = orcl_get_published(qp_name, (void **) &ret);

	/* 
	 * The registry lookup returns only if we get a unique QP for @qp_name, or
	 * if the memcached lookup succeeds but we don't have an entry for @qp_name.
	 */
	assert(ret_len == sizeof(struct orcl_qp_attr) || ret_len == -1);
	_unused(ret_len);

	return ret;
}

void get_all_published_qp(struct orcl_context* orcl_ctx){
    // 首先获取所有normal rc的远程信息
    int i;
    for (i = 0; i < orcl_ctx->group_size; i++){
        // 跳过自己
        // 不需要与自己建立连接
        int inner_group_index = orcl_ctx->node_id % orcl_ctx->group_size;

        if(inner_group_index == i){
            continue;
        }

        char remote_qp_name[ORCL_QP_NAME_SIZE];

        int remote_node_id = orcl_ctx->group_size * orcl_ctx->group_id + i;
        
        sprintf(remote_qp_name, "server-normal-%d-%d-%d", remote_node_id, orcl_ctx->orcl_ctx_id, inner_group_index);

        // 获取对应远程节点的qp信息
        struct orcl_qp_attr *remote_qp = NULL;

        while (remote_qp == NULL)
        {
            remote_qp = orcl_get_published_qp(remote_qp_name);
            if (remote_qp == NULL)
            {
                usleep(200000);
            }
        }

        // 初始化对应的远程信息
        memcpy(&(orcl_ctx->others_normal_qp[i]), remote_qp, sizeof(struct orcl_qp_attr));        
    }

    // 然后获取所有forward rc
    for (i = 0; i < orcl_ctx->group_size; i++){
        // 跳过自己
        // 不需要与自己建立连接
        int inner_group_index = orcl_ctx->node_id % orcl_ctx->group_size;

        if(inner_group_index == i){
            continue;
        }

        char remote_qp_name[ORCL_QP_NAME_SIZE];

        int remote_node_id = orcl_ctx->group_size * orcl_ctx->group_id + i;
        
        sprintf(remote_qp_name, "server-forward-%d-%d-%d", remote_node_id, orcl_ctx->orcl_ctx_id, inner_group_index);

        // 获取对应远程节点的qp信息
        struct orcl_qp_attr *remote_qp = NULL;

        while (remote_qp == NULL)
        {
            remote_qp = orcl_get_published_qp(remote_qp_name);
            if (remote_qp == NULL)
            {
                usleep(200000);
            }
        }

        // 初始化对应的远程信息
        memcpy(&(orcl_ctx->others_forward_qp[i]), remote_qp, sizeof(struct orcl_qp_attr));        
    }
}

// 获取其他节点的所有dct
void sync_get_all_dc_target(struct orcl_context_sync* sync_ctx){
    // 获取当前上下文所需要的所有dct
    assert(sync_ctx != NULL);

    // 遍历所有节点，当前节点除外
    int i;
    for(i = 0; i < sync_ctx->node_num; i++){
        if(sync_ctx->node_id == i){
            // 当前组不需要转发问题，所以不需要知道转发节点的信息
            continue;
        }

        char remote_qp_name[ORCL_QP_NAME_SIZE];

        // 第一个是转发节点所在的组号，第二个是转发节点的组内id，第三个当前节点所在组，第四个是上下文编号
        sprintf(remote_qp_name, "server-sync-dct-%d-%d", i, sync_ctx->orcl_ctx_id);

        struct orcl_dct_attr *remote_dct = NULL;

        while (remote_dct == NULL)
        {
            remote_dct = orcl_get_published_dct(remote_qp_name);
            if (remote_dct == NULL)
            {
                usleep(200000);
            }
        }

        // 拷贝
        memcpy(&(sync_ctx->others_dct[i]), remote_dct, sizeof(struct orcl_dct_attr));

        // 建立对应的路由信息
        struct ibv_ah_attr ah_attr;
        memset((void *) &ah_attr, 0, sizeof(ah_attr));

        ah_attr.is_global = 0;
        ah_attr.dlid = remote_dct->lid;
        ah_attr.sl = 0;
        ah_attr.src_path_bits = 0;
        // 当前上下文使用的端口
        ah_attr.port_num = sync_ctx->dev_port_id;

        // 根据编译参数，来决定是不是使用global的方式来发送
#if USE_GID_ROUTE == 1
        ah_attr.is_global = 1;
        ah_attr.grh.dgid = remote_dct->gid;
        ah_attr.grh.sgid_index = 0;
        ah_attr.grh.hop_limit = 128;
#endif

        // 分别为两个dci创建路由信息
        int j;
        for(j = 0; j < sync_ctx->dci_num; j++){
            sync_ctx->other_dct_ah_attr[j][i] = ibv_create_ah(sync_ctx->dci_pd[j], &ah_attr);
        }
    }
}

void orcl_get_all_dc_target(struct orcl_context* orcl_ctx){
    // 获取当前上下文所需要的所有dct
    assert(orcl_ctx != NULL);

    // 当前节点的组内索引
    int inner_group_index = orcl_ctx->node_id % orcl_ctx->group_size;

    // 遍历所有组，当前组除外
    int i;
    for(i = 0; i < orcl_ctx->group_num; i++){
        if(orcl_ctx->group_id == i){
            // 当前组不需要转发问题，所以不需要知道转发节点的信息
            continue;
        }

        char remote_qp_name[ORCL_QP_NAME_SIZE];

        // 第一个是转发节点所在的组号，第二个是转发节点的组内id，第三个当前节点所在组，第四个是上下文编号
        sprintf(remote_qp_name, "server-dct-%d-%d-%d-%d", i, inner_group_index, orcl_ctx->group_id, orcl_ctx->orcl_ctx_id);

        struct orcl_dct_attr *remote_dct = NULL;

        while (remote_dct == NULL)
        {
            remote_dct = orcl_get_published_dct(remote_qp_name);
            if (remote_dct == NULL)
            {
                usleep(200000);
            }
        }

        memcpy(&(orcl_ctx->others_dct[i]), remote_dct, sizeof(struct orcl_dct_attr));

        // 建立对应的路由信息
        struct ibv_ah_attr ah_attr;
        memset((void *) &ah_attr, 0, sizeof(ah_attr));
        ah_attr.is_global = 0;
        ah_attr.dlid = remote_dct->lid;
        ah_attr.sl = 0;
        ah_attr.src_path_bits = 0;
        // 当前上下文使用的端口
        ah_attr.port_num = orcl_ctx->dev_port_id;

        // 分别为两个dci创建路由信息
        int j;
        for(j = 0; j < orcl_ctx->dci_num; j++){
            orcl_ctx->other_dct_ah_attr[j][i] = ibv_create_ah(orcl_ctx->dci_pd[j], &ah_attr);
        }
    }
}
